package LogServer.kafka

import LogServer.constants.PropertiesZookeeper
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka.KafkaUtils

/**
  * Created by root on 17-2-14.
  */
class KafkaConnector {
  /**
    * 创建Kafka连接
    * @param ssc
    * @return
    */
  def createStream(ssc: StreamingContext): DStream[String] ={
    val topicMap = Map("test" -> 1)
    //zookeeper quorums server list
    val zkQuorum = PropertiesZookeeper.getZkQuorum()
    //consumer group
    val group = "test-group"
    val dstream = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2)
    dstream
  }
}
